#include <cserver/db.h>
#include <cserver/context.h>
#include <cserver/data.h>
#include <boost/log/trivial.hpp>
#include <iomanip>
#include <iostream>
#include <exception>
#include <thread>
#include <vector>
#include <atomic>
#include <mutex>
#include <cstdlib>
#include <cmath>
#include <bitset>


#include <condition_variable>

#include <pqxx/connection.hxx>
#include <pqxx/transaction.hxx>
#include <boost/format.hpp>
#include <boost/lexical_cast.hpp>
namespace csrv{
	void unused(int){}
	struct SendSMS{
		Context* context;
		SendSMS(Context& ctx, std::chrono::system_clock::time_point tp, std::chrono::system_clock::duration d) 
			: context(&ctx)
			  , point(tp), duration(d)
		{}

		SendSMS(const SendSMS&) = default;
		
		void operator()() {
			context->database->sendHydrantSMS();

			if (duration > std::chrono::seconds(0)){
				point += duration;
				context->timer->append(point, *this);
			}
		}

		std::chrono::system_clock::time_point point;
		std::chrono::system_clock::duration duration;
	};


	class DatabaseImp{
		public:
			Context& context;
			const Config& config;
			pqxx::connection dbconn;
			std::thread worker;
			std::atomic<bool> workerIsRunning = {false};
			using action_type = std::function<void(pqxx::work&)>;
			std::mutex mutex;
			std::vector<action_type> queue;
			std::condition_variable condition_variable;

			DatabaseImp(Context& ctx) 
				: context(ctx)
				, config(ctx.config)
				, dbconn(ctx.config.db_string.c_str())
			{
				dbconn.prepare("insert_device", "INSERT INTO t_device(moteeui, gweui, seqno, rfchan, chan, freq, modu, datr, adr, sf, rssi, snr, "
						"light_pwm,voltage,current,power,energy,x_axis,y_axis,z_axis,als,lcm)"
						" values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22)");
				dbconn.prepare("write_test", "INSERT INTO test(text) values($1)");
				dbconn.prepare("insert_node_data", 
						"insert into t_node_data(euid, log_time, rtime, rtime_from_gw, gweuid, seqno, light_pwm, voltage, current, power, energy) "
						"values($1, CURRENT_TIMESTAMP, $2, $3, $4, $5, $6, $7, $8, $9, $10)");
				dbconn.prepare("write_soak",
						R"( insert into t_soak_data(euid, time, signal, sim_id, voltage, soaking_status)
						select t_node_info2.id, now(), $2, $3, $4, $5 
						from t_node_info2 where t_node_info2.euid=$1
						)");
				dbconn.prepare("write_hydrant", "select append_hydrant_data($1, $2, $3, int2($4), $5, $6)");
				dbconn.prepare("write_well_lip", "select append_well_lip_data(text($1), int2($2), text($3), int2($4), int2($5))");
				dbconn.prepare("write_fire_door", "insert into t_firedoor_data(euid, time, signal, sim_id, warning_status, voltage) " 
						" select t_node_info2.id, localtimestamp, int2($2), $3, $4, $5 from t_node_info2 where t_node_info2.euid=$1");
				dbconn.prepare("write_sprinkler", "insert into t_sprinkler_data(euid, time, signal, simid, pressure, voltage) " 
						" select t_node_info2.id, localtimestamp, int2($2), $3, $4, $5 from t_node_info2 where t_node_info2.euid=$1");
				dbconn.prepare("write_fire_box", "insert into t_firebox_data(euid, time, signal, simid, depth, voltage) " 
						" select t_node_info2.id, localtimestamp, int2($2), $3, $4, $5 from t_node_info2 where t_node_info2.euid=$1");
				dbconn.prepare("write_fire_pump", "insert into t_firepump_data(euid, time, signal, sim_id, pressure, voltage) " 
						" select t_node_info2.id, localtimestamp, int2($2), $3, $4, $5 from t_node_info2 where t_node_info2.euid=$1");
				dbconn.prepare("write_water_quality", "insert into t_waterquality_data(euid, time, signal, sim_id, temperature,"
						" turbidity, ph, orp, conductivity, ammonia_nitrogen, oxygen, enclosure, latitude, longitude, uvt, testdata,"
						" turbidity_a, turbidity_b, ph_a, ph_b, orp_a, orp_b, oxygen_a, oxygen_b, conductivity_a, conductivity_b, "
						" ammonia_nitrogen_a, ammonia_nitrogen_b, cod_a, cod_b, uvt_a, uvt_b "
						"  ) " 
						" select t_node_info2.id, localtimestamp, int2($2), $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, t_node_info2.ischecking, "
						" a.turbidity_a, a.turbidity_b, a.ph_a, a.ph_b, a.orp_a, a.orp_b, a.oxygen_a, a.oxygen_b, a.conductivity_a, a.conductivity_b, "
						" a.ammonia_nitrogen_a, a.ammonia_nitrogen_b, a.cod_a, a.cod_b, a.uvt_a, a.uvt_b "
						" from t_node_info2 left join t_waterquality_coefficient a on t_node_info2.id = a.euid where t_node_info2.euid=$1");

				dbconn.prepare("write_environment", "select append_environment($1, int2($2), $3, $4,$5,$6,$7,$8,$9,$10,$11,$12)");
				dbconn.prepare("update_simid", "update t_node_info2 set (sim_id, last_ip_port) = ($2, $3) where euid = $1");
				dbconn.prepare("get_environment_inteval", "update  t_environment_node as a set is_report = 1 where a.is_report = 0 and "
						" exists (select 1 from t_node_info2 b where b.id = a.euid and b.euid = $1) RETURNING a.interval");
				dbconn.prepare("get_sms_today", "select count(*) from t_sms_log where date(time) = current_date");
				dbconn.prepare("log_sms", "insert into t_sms_log (time) values(now())");
				dbconn.prepare("write_voiceHB", "insert into t_horn_data(euid, time, signal, simid, status)"
						" select t_node_info2.id, localtimestamp, int2($2), $3, $4 from t_node_info2 where t_node_info2.euid=$1");
				dbconn.prepare("get_voice_id", "select * from (select command_type from t_horn_command_data where euid in (select id from t_node_info2 where euid=$1)" 
						" order by time desc limit 1) a where a.command_type is not null and a.command_type <> $2");
				dbconn.prepare("get_forward_endpoint", "select a.orgip, a.port from t_organization as a, t_node_info2 as b where a.id = b.orgid and b.euid = $1");


				dbconn.prepare("get_phone_number", "select orgid, phone_number from t_sms_user where phone_number is not null");
				dbconn.prepare("get_timers", "select start, repeat from t_timer_info");
				dbconn.prepare("hydrant_status_summary", "select status, count(*) from (select (case when pressure>=100000 then 5"
					" when pressure between 50000 and 99999 then 4"
					" when pressure between  10000 and 49999 then 1 "
					" when pressure < 10000 then 2"
					" else 3  end"
					" ) as status from v_hydrant_status2 where orgid=$1) as a group by status order by status");

				dbconn.prepare("clear_device_init_flag", "update t_electric_flag set down_flag=0 where euid in (select id from t_node_info2 where euid = $1)");
				dbconn.prepare("clear_device_param_flag", "update t_electric_flag set param_flag=0 where euid in (select id from t_node_info2 where euid = $1)");
				dbconn.prepare("write_electric", 
						R"(insert into t_electric_data(euid, time, signal, sim_id, 
						 "voltage_A", "voltage_B", "voltage_C", "current_A", "current_B", "current_C",  
						 "phaseangle_A", "phaseangle_B", "phaseangle_C",
						 frequency, power_factor, 
						 "ThreePhase_ActivePower", "ThreePhase_ReActivePower", "ThreePhase_Apparent_Power", 
						 "Fundamental_Wave_Residual_Current", "Full_Wave_Residual_Current",
						 in_status, out_status, 
						 "temperature_A", "temperature_B", "temperature_C", "temperature_D",
						 voltagewire_selftest, currentwire_selftest, temperaturewire_selftest, 
						 totalharmonic_active_power, 
						 "Forward_Active_Energy", "Forward_Reactive_Energy", "Reverse_Active_Energy", "Reverse_Reactive_Energy", 
						 "currentTHD_A", "currentTHD_B", "currentTHD_C",
						 "voltageTHD_A", "voltageTHD_B", "voltageTHD_C",
						 "Voltage_Imbalance", "Current_Imbalances", "warning_Code"
						 ) select t_node_info2.id, localtimestamp, int2($2), $3, $4, $5, $6, $7, $8, $9,
						 $10, $11, $12, $13, $14, $15, $16, $17, $18, $19,
						 $20, $21, $22, $23, $24, $25, $26, $27, $28, $29,
						 $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, 
						 $40, $41, $42
						 from t_node_info2 where t_node_info2.euid=$1)");
				dbconn.prepare("get_electric_flags", "select down_flag, param_flag from  t_electric_flag where euid in (select id from t_node_info2 where euid = $1)");
				dbconn.prepare("get_electric_down", "select voltage_rating_firstside, voltage_rating_secondside, "
						" current_rating_firstside, current_rating_secondside, ct_type, wiringtype, do_value, erasurestatus "
						" from  t_electric_down where euid in (select id from t_node_info2 where euid = $1)");
				dbconn.prepare("get_electric_param", "select mode, key, upper_limit, lower_limit, action_delay, return_delay, action1, action2 "
						" from  t_electric_param where euid in (select id from t_node_info2 where euid = $1)");
				dbconn.prepare("save_xhs_analog", 
						R"( select update_xhs_electric_data($1::boolean, $2, $3, $4, $5::double precision, $6, $7)
						)");
				dbconn.prepare("save_xhs_analog_config", 
						R"(select update_xhs_electric_node($1::boolean, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
						)");
				dbconn.prepare("get_xhs_flags", "select dtu_flag, down_flag from  t_xhs_electric_flag where euid in (select id from t_node_info2 where euid = $1)");
				dbconn.prepare("get_xhs_down", "select channel_number, operating_code, address from  t_xhs_electric_down where euid in (select id from t_node_info2 where euid = $1)");
				dbconn.prepare("get_xhs_dtu", "select signal_type, normal_heartbeat,warn_heartbeat, warn_heartbeat_duration, error_range  "
						" from  t_xhs_electric_dtu where euid in (select id from t_node_info2 where euid = $1)");
				dbconn.prepare("clear_xhs_flags", "update t_xhs_electric_flag set  dtu_flag = 0, down_flag = 0 where euid in (select id from t_node_info2 where euid = $1)");

				dbconn.prepare("write_fire_facility_sys_status",
						R"( insert into t_fire_facility_sys_status(euid, time, sequence_id, data_time, src_addr, dest_addr, system_type, system_addr, status)
						select t_node_info2.id, now(), $2, $3, $4, $5, $6, $7, $8
						from t_node_info2 where t_node_info2.euid=$1
						)");
				dbconn.prepare("update_fire_facility_sys_status_state",
						R"(  select update_fire_facility_sys_status_state(text($1), $2::integer, $3::timestamp, text($4), $5::smallint, $6::smallint, $7::bit(16), $8::boolean))");

				dbconn.prepare("write_fire_facility_part_status",
						R"( insert into t_fire_facility_part_status(euid, time, sequence_id, data_time, src_addr, dest_addr, system_type, system_addr, 
						part_type, part_area, part_number, status, part_desc)
						select t_node_info2.id, now(), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12
						from t_node_info2 where t_node_info2.euid=$1
						)");
				dbconn.prepare("update_fire_facility_part_status_state",
						R"(  select update_t_fire_facility_part_status_state(text($1), $2::integer, $3::timestamp, text($4), $5::smallint, $6::smallint, 
							$7::smallint, $8::smallint, $9::smallint,
							$10::bit(16), text($11), $12::boolean)
						)");

				dbconn.prepare("write_fire_device_run_status",
						R"( insert into t_fire_device_run_status(euid, time, sequence_id, data_time, src_addr, dest_addr, status)
						select t_node_info2.id, now(), $2, $3, $4, $5, $6
						from t_node_info2 where t_node_info2.euid=$1
						)");
				dbconn.prepare("update_fire_device_run_status_state",
						R"( insert into t_fire_device_run_status_state(euid, time, sequence_id, data_time, src_addr, dest_addr, status)
						select t_node_info2.id, now(), $2, $3, $4, $5, $6
						from t_node_info2 where t_node_info2.euid=$1
						on conflict (euid, src_addr)
						do update set status = $6
						)");

				dbconn.prepare("write_fire_device_op_info",
						R"( insert into t_fire_device_op_info(euid, time, sequence_id, data_time, src_addr, dest_addr, operator_id, status)
						select t_node_info2.id, now(), $2, $3, $4, $5, $6, $7
						from t_node_info2 where t_node_info2.euid=$1
						)");

				dbconn.prepare("update_fire_device_op_info_state",
						R"( insert into t_fire_device_op_info_state(euid, time, sequence_id, data_time, src_addr, dest_addr, operator_id, status)
						select t_node_info2.id, now(), $2, $3, $4, $5, $6, $7
						from t_node_info2 where t_node_info2.euid=$1
						on conflict (euid, src_addr)
						do update set status = $7
						)");
						
				dbconn.prepare("save_fire_pump_info",
						R"( insert into t_fire_pump_data(euid, time, signal, sim_id, voltage, status)
						select t_node_info2.id, now(), $2, $3, $4, $5
						from t_node_info2 where t_node_info2.euid=$1
						)");
				dbconn.prepare("save_draught_fan_info",
						R"( insert into t_draught_fan_data(euid, time, signal, sim_id, voltage, status)
						select t_node_info2.id, now(), $2, $3, $4, $5
						from t_node_info2 where t_node_info2.euid=$1
						)");
				dbconn.prepare("save_waterater_data",
						R"( insert into t_waterater_consumption_data(euid, time, signal, sim_id, voltage, fluid_velocity, positive_cumulative_flow,
							negative_cumulative_flow, net_cumulative_flow, error_code, work_steps, upstream_signal, downstream_signal)
						select t_node_info2.id, now(), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12
						from t_node_info2 where t_node_info2.euid=$1
						)");

				dbconn.prepare("save_spongecity_data",
						R"( select save_spongecity_data(text($1), $2::integer, text($3), $4::integer, 
						$5::integer, $6::integer, $7::integer, $8::integer,
						$9::integer, $10::integer, $11::integer, $12::integer,
						$13::integer, $14::integer, $15::integer, $16::integer,
						$17::integer, $18::integer, $19::double precision, $20::integer,
						$21::integer, $22::integer, $23::integer, $24::integer
						)
						)");

				dbconn.prepare("save_spongecity_upconfig",
						R"( select save_spongecity_upconfig(text($1), 
						$2::smallint, $3::smallint, $4::smallint, $5::smallint,
						$6::smallint, $7::smallint, $8::smallint, $9::smallint, $10::smallint
						)
						)");
				dbconn.prepare("get_spongecity_downconfig",
						R"( select t_spongecity_downconfig.status, rain_wakeup, rain_report, norain_wakeup, norain_report, flow_use, ss_use,
						cod_use, rain_use, flow_pipe, flow_height, cod_interval, cod_circle, flow_cal_con
						from t_spongecity_downconfig, t_node_info2 
						where t_node_info2.euid = $1 and t_node_info2.id = t_spongecity_downconfig.euid
						)");
				dbconn.prepare("clear_spongecity_downconfig",
						R"( update t_spongecity_downconfig set status= 0 
						where euid in (select id from t_node_info2 where euid = $1)
						)");


				pqxx::work w(dbconn);
				auto result = w.prepared("get_timers").exec();

				using namespace std::chrono;
				typedef duration<int, std::ratio<60*60*24>> days_type;
				auto today = time_point_cast<days_type>(system_clock::now());
				
				for(auto row : result){
					auto itr = row.begin();
					auto point = today + std::chrono::seconds(itr->as<int>());
					auto duration = std::chrono::seconds((++itr)->as<int>());
					while (point < system_clock::now())
						point += duration;
					context.timer->append(point, SendSMS(context, point, duration));
				}
			}
			void writeTest(const std::string& str){
				std::unique_lock<std::mutex> lock(mutex);
				queue.emplace_back([=](pqxx::work& w){
						w.prepared("write_test")(str).exec();
				});
				condition_variable.notify_all();
			}

			void saveDeviceRecord(const app_t& app){
				if (!app.userdata || app.gwrx.empty()){
					BOOST_LOG_TRIVIAL(warning) <<"No userdata.";
					return;
				}

				auto& udata = *app.userdata;
				std::vector<uint8_t> buf;
				if (!base64_decode(udata.payload, buf)){
					BOOST_LOG_TRIVIAL(warning) <<"Bad base64 encoded payload: " << udata.payload;
					return;
				}

				std::stringstream sstr;
				sstr << "Payload (size=" << buf.size() << "):[";
				for(auto ch : buf)
					sstr << ' ' << std::setfill('0') << std::setw(2) << std::hex << uint32_t(ch);
				sstr << " ]";

				if (buf.empty()) {
					return;
				}
				BOOST_LOG_TRIVIAL(info) << sstr.str();
				if(*buf.begin() != 0x02){
					BOOST_LOG_TRIVIAL(warning) << "Unknown command."; 
					return;
				}

				node_info_t node;
				if(!parse_payload(buf, node)){
					BOOST_LOG_TRIVIAL(debug) << "Failed to parse payload";
					return;
				}

				auto& gwrx = app.gwrx[0];
				try{
					BOOST_LOG_TRIVIAL(info) 
						<< " time:" << gwrx.time 
						<< " moteeui:" << (app.moteeui) << "(" << std::setfill('0') << std::setw(16) << std::hex << app.moteeui <<")" << std::dec
						<< " gwrx.eui:" << (gwrx.eui) << "(" << std::setfill('0') << std::setw(16) << std::hex << gwrx.eui << ")" << std::dec
						<< " seqno:" << (udata.seqno)
						<< " rfch:" << (gwrx.rfch) << " chan:" << (gwrx.chan)
						<< " freq:" << (udata.motetx.freq) << " modu:" << (udata.motetx.modu)
						<< " datr:" << (udata.motetx.datr)  << " adr:" << (udata.motetx.adr)
						<< " port:" << (udata.port)
						<< " rssi:" << (gwrx.rssi) << " lsnr:" << (gwrx.lsnr)
						<< " light_pwm:" << (int32_t(node.light_pwm)) << " voltage:" << (node.voltage)
						<< " current:" << (node.current) << " power:" << (node.power)
						<< " energy:" << (node.energy) << " x_axis:" << (node.x_axis)
						<< " y_axis:" << (node.y_axis) << " z_axis:" << (node.z_axis)
						<< " als:" << (node.als) << " lcm:" << (int32_t(node.lcm));
#if 1
					std::unique_lock<std::mutex> lock(mutex);
					queue.emplace_back([=](pqxx::work& w){
							w.prepared("insert_node_data")
							(app.moteeui)
							(gwrx.time)
							(gwrx.timefromgateway)
							(gwrx.eui)
							(udata.seqno)
							(int32_t(node.light_pwm))
							(node.voltage)
							(node.current)
							(node.power)
							(node.energy)
							.exec();
							});
					condition_variable.notify_all();
#endif
				}
				catch(std::exception& e){
					BOOST_LOG_TRIVIAL(warning) << "Failed to insert db: " << e.what();
				}
				catch(...){
					BOOST_LOG_TRIVIAL(warning) << "Failed to insert db";
				}
			}
			bool parse_payload(const std::vector<uint8_t>& data, node_info_t& node){
				auto end = data.end();
				for (auto itr =  data.begin(); itr != end;){
					auto cmd = *itr;
					if (cmd != 0x02 || ++itr == end) break;
					auto len = *itr++;
					if (len > (end - itr))	break;	// no enough value data

					parse_node_info(&*itr, len, node);
					return true;
				}
				return false;
			}

			void print_node(const node_info_t& node){
				std::stringstream sstr;
				sstr << "node_info:{";

				sstr 
					<< " light_pwm:" << uint32_t(node.light_pwm)
					<< " voltage:" << uint32_t(node.voltage)
					<< " current:" << uint32_t(node.current)
					<< " power:" << uint32_t(node.power)
					<< " energy:" << uint32_t(node.energy)
					<< " x-axis:" << int32_t(node.x_axis)
					<< " y-axis:" << int32_t(node.y_axis)
					<< " z-axis:" << int32_t(node.z_axis)
					<< " als:" << int32_t(node.als)
					<< " lcm:" << int32_t(node.lcm)

					<< " }";
				BOOST_LOG_TRIVIAL(info) << sstr.str();
			}

			void do_work(){
				std::unique_lock<std::mutex> lock(mutex);
				while(workerIsRunning){
					if (queue.empty())
						condition_variable.wait(lock);
					if (queue.empty())
						continue;
					try{
						std::vector<action_type> actions;
						actions.swap(queue);
						pqxx::work w(dbconn);
						for(auto& fun : actions)
							fun(w);
						w.commit();
					}
					catch(std::exception&e){
						BOOST_LOG_TRIVIAL(warning) << "Database Exception: " << e.what();
					}
					catch(...){
						BOOST_LOG_TRIVIAL(warning) << "Unknown Exception";
					}
				}
			}
			void sendHydrantSMS(){
				std::unique_lock<std::mutex> lock(mutex);
				queue.emplace_back([=](pqxx::work& w){
						std::map<std::string, std::vector<std::string>> users;
						for (auto row :  w.prepared("get_phone_number").exec()){
							auto field = row.begin();
							std::string orgid = field->c_str(); ++field;
							users[orgid].push_back(field->c_str());
						}

						for (auto& v : users){
							std::stringstream phone;
							bool first = true;
							for(auto& number : v.second){
								if (first) first = false;
								else phone << ",";
								phone << number;
							}
							std::map<std::string, std::string> data;
							for(int i = 1; i < 6; ++i)
								data[boost::lexical_cast<std::string>(i)]="0";
							auto result = w.prepared("hydrant_status_summary")(v.first).exec();
							for (auto row : result){
								auto itr = row.begin();
								std::string status = itr->c_str();
								data[status] =(++itr)->c_str();
							}
							auto fmt = boost::format(config.send_sms);
							for(auto& v: data)
								fmt % v.second;	// args 1~5
							fmt % phone.str();	// arg 6

							auto cmd = str(fmt);
							auto ret = system(cmd.c_str());
							unused(ret);
						}
				});
				condition_variable.notify_all();
			}
	};

	Database::Database(Context& ctx)
	{
		try{
			m_imp.reset(new DatabaseImp(ctx));
		} catch(std::exception& e){
			std::cout << "Database ctor: " << e.what() << std::endl;
			throw;

		}
	}
	Database::~Database() = default;

	void Database::saveDeviceRecord(const app_t& app){
		m_imp->saveDeviceRecord(app);
	}
	void Database::writeTest(const std::string& str){
		m_imp->writeTest(str);
	}
	void Database::sendHydrantSMS(){
		m_imp->sendHydrantSMS();
	}

	void Database::update_sim_id(const std::string& euid, const std::string& simid, const std::string& ipaddr){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("update_simid")(euid)(simid)(ipaddr).exec();
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::saverHhydrant(const std::string& ip_addr_port, const std::string& euid, int32_t pressure, int32_t trend, 
			uint16_t signal, const std::string& sim_id, int32_t voltage){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("update_simid")(euid)(sim_id)(ip_addr_port).exec();
				auto old_pressure = w.prepared("write_hydrant")
				(euid)(pressure)(trend)(signal)(sim_id)(voltage).exec().begin()->begin()->as<int>(0);
				if ( old_pressure > 100000 && pressure < 100000){
				auto send_count = w.prepared("get_sms_today").exec().begin()->begin()->as<int>(0);
				if (send_count < m_imp->config.sms_limit){
				std::stringstream text;
				try{
				text << boost::format(m_imp->config.send_sms) % euid % m_imp->config.mobile_numbers;
				auto ret = system(text.str().c_str());
				unused(ret);
				w.prepared("log_sms").exec();
				}catch(...){}
				}
				}
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::saverSprinkler(const std::string& ip_addr_port, const std::string& euid, int32_t pressure, int32_t voltage, uint16_t signal, const std::string& sim_id){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("write_sprinkler") (euid)(signal)(sim_id)(pressure)(voltage).exec();
				w.prepared("update_simid")(euid)(sim_id)(ip_addr_port).exec();
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::saverFirebox(const std::string& ip_addr_port, const std::string& euid, int32_t water_deepth, int32_t voltage, uint16_t signal, const std::string& sim_id){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("write_fire_box") (euid)(signal)(sim_id)(water_deepth)(voltage).exec();
				w.prepared("update_simid")(euid)(sim_id)(ip_addr_port).exec();
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::saverFirePump(const std::string& ip_addr_port, const std::string& euid, int32_t pressure, int32_t voltage, uint16_t signal, const std::string& sim_id){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("write_fire_pump") (euid)(signal)(sim_id)(pressure)(voltage).exec();
				w.prepared("update_simid")(euid)(sim_id)(ip_addr_port).exec();
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::saveFiredoor(const std::string& ip_addr_port, const std::string& euid, uint16_t signal, const std::string& sim_id, int32_t voltage, uint16_t warning_status, uint16_t /*submerged*/){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("write_fire_door") (euid)(signal)(sim_id)(warning_status)(voltage).exec();
				w.prepared("update_simid")(euid)(sim_id)(ip_addr_port).exec();
				});
		m_imp->condition_variable.notify_all();

	}
	void Database::saveWellLip(const std::string& ip_addr_port, const std::string& euid, uint16_t signal, const std::string& sim_id, uint16_t cover_status, uint16_t submerged){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("write_well_lip") (euid)(signal)(sim_id)(cover_status)(submerged).exec();
				w.prepared("update_simid")(euid)(sim_id)(ip_addr_port).exec();
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::saverWaterQuality(const std::string& ip_addr_port, const std::string& euid, uint16_t signal, const std::string& sim_id
			, uint32_t temperature, uint32_t ntu, float ph, float orp, float ec, float nh4, uint32_t do_value, int16_t mover,  float latitude, float longitude, uint32_t uvt){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("write_water_quality") (euid)(signal)(sim_id)(temperature)(ntu)
				(ph, !std::isnan(ph))(orp, !std::isnan(orp))
				(ec, !std::isnan(ec))(nh4, !std::isnan(nh4))(do_value)(mover)
				(latitude, !std::isnan(latitude))(longitude, !std::isnan(longitude))(uvt).exec();
				w.prepared("update_simid")(euid)(sim_id)(ip_addr_port).exec();
				});
		m_imp->condition_variable.notify_all();

	}
	void Database::saveEnvironment(const std::string& ip_addr_port, const std::string& euid, uint16_t signal, const std::string& sim_id, uint16_t interval, const std::vector<Environment>& env_data,
			std::function<void(uint16_t interval)> sender){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
#if 0
				uint16_t adjust_second = 0;
				for(auto& e : env_data){
					w.prepared("write_environment") (euid)(signal)(sim_id)(adjust_second)(e.temperature)(e.humidity)(e.pm25)(e.pm10)(e.pm1)
						(e.wind_speed)(e.wind_direction)(e.noise).exec();
					adjust_second += interval;
				}
				w.prepared("update_simid") (euid)(sim_id)(ip_addr_port).exec();
#endif
				auto result = w.prepared("get_environment_inteval") (euid).exec();
				if (!result.empty()){
					BOOST_LOG_TRIVIAL(info) << "DEBUG send inteval: " << euid << ", " << interval << ", count=" << env_data.size();
					uint16_t new_interval = uint16_t(result.begin()->begin()->as<int>());
					sender(new_interval);
				}
				else 
					sender(interval);
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::save_electric(const std::string& ip_addr_port, const pcm_350a_data& d, std::function<void(const std::vector<char>&)> sender){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
#if 0
				w.prepared("write_electric")(d.euid)(uint16_t(d.signal))(d.simid)
				(d.meter_voltage.a)(d.meter_voltage.b)(d.meter_voltage.c)
				(d.meter_current.a) (d.meter_current.b) (d.meter_current.c)
				(d.meter_current_phase.a) (d.meter_current_phase.b) (d.meter_current_phase.c)
				(d.frequency)(d.power_factor)
				(d.meter_3phase_power.active) (d.meter_3phase_power.reactive) (d.meter_3phase_power.apparent)
				(d.base_residual_current)(d.fullwave_residual_current)
				(d.di_value)(d.do_value)
				(d.meter_temperature.a) (d.meter_temperature.b) (d.meter_temperature.c) (d.meter_temperature.d)
				(d.meter_self_test.voltage_wire)(d.meter_self_test.current_wire)(d.meter_self_test.temperature_wire)
				(d.total_harmonic_active_power)
				(d.meter_energy.forward_active) (d.meter_energy.forward_reactive) (d.meter_energy.reverse_active) (d.meter_energy.reverse_reactive)
				(d.meter_current_thd.a) (d.meter_current_thd.b) (d.meter_current_thd.c)
				(d.meter_voltage_thd.a) (d.meter_voltage_thd.b) (d.meter_voltage_thd.c)
				(d.unbalance_factor.voltage)(d.unbalance_factor.current)
				(d.warning_code)
				.exec();
				w.prepared("update_simid") (d.euid)(d.simid)(ip_addr_port).exec();
#endif

				auto result = w.prepared("get_electric_flags")(d.euid).exec();
				if (!result.empty()){
					auto itr = result.begin()->begin();
					int init_flag = itr->as<int>(0);
					++itr;
					int param_flag = itr->as<int>(0);
					if (init_flag != 0){
						auto res = w.prepared("get_electric_down")(d.euid).exec();
						if (!res.empty()){
							const auto& record = *res.begin();
							pt_ct_init_data init_data{};
							init_data.voltage_rating_first_side  = record[0].as<float>(0);
							init_data.voltage_rating_second_side = record[1].as<float>(0);
							init_data.current_rating_first_side = record[2].as<float>(0);
							init_data.current_rating_second_side = record[3].as<float>(0);
							uint16_t ct_type = record[4].as<uint16_t>(0);
							uint16_t wiring_type = record[5].as<uint16_t>(0);
							uint16_t do_value = record[6].as<uint16_t>(0);
							uint8_t warning_void_ctrl = record[7].as<uint16_t>(0);

							std::vector<char> response;
							buffer_out wr(response);
							auto ar = s11n::as_sink(wr);
							ar & K_signature & K_version & K_reserved 
								& uint16_t(0xe)
								& uint16_t(sizeof(init_data) + 17)
								& uint16_t(0x114)
								& init_data
								& uint16_t(0x115)	// ct_type
								& ct_type
								& uint16_t(0x116)
								& wiring_type
								& uint16_t(0x117)
								& do_value
								& uint16_t(0x11c)
								& warning_void_ctrl
								;
							sender(response);
						}
					}
					if (param_flag != 0){
						auto dataset = w.prepared("get_electric_param")(d.euid).exec();
						std::vector<pcm_threshold_setting> params;
						for( auto record : dataset){
							pcm_threshold_setting setting{};
							setting.triger_type = record[0].as<uint16_t>(0);
							setting.monitor_object = record[1].as<uint16_t>(0);
							setting.triger_threshold_up = record[2].as<float>(0);
							setting.triger_threshold_down = record[3].as<float>(0);
							setting.act_delay = record[4].as<uint16_t>(0);
							setting.return_delay = record[5].as<uint16_t>(0);
							setting.triger_act1 = record[6].as<uint16_t>(0);
							setting.triger_act2 = record[7].as<uint16_t>(0);
							params.push_back(setting);
						}
						std::vector<char> res2;
						buffer_out wr2(res2);
						auto ar2 = s11n::as_sink(wr2);
						ar2 & K_signature & K_version & K_reserved 
							& uint16_t(0xf)
							& uint16_t(sizeof(pcm_threshold_setting)* params.size() + 3)
							& uint16_t(0x119)
							& uint8_t(params.size());
						for(auto&v : params)
							ar2 & v;
						BOOST_LOG_TRIVIAL(debug) << "DEBUG send threshold: " << params.size() << " bytes " << res2.size();
						sender(res2);
					}
				}

		});
		m_imp->condition_variable.notify_all();

	}

	static void check_xhs_flags(pqxx::work& w, const std::string& euid, const std::function<void(const std::string&)>& sender){
		auto result = w.prepared("get_xhs_flags")(euid).exec();
		if (!result.empty()){
			auto itr = result.begin()->begin();
			int dtu_flag = itr->as<int>(0);
			++itr;
			int down_flag = itr->as<int>(0);

			if (down_flag != 0){
				auto res = w.prepared("get_xhs_down")(euid).exec();
				std::stringstream sstr;
				sstr << "@@" << 4 // message type
					<< "|" << 1024 // message sequence number, hard coded, FIXME
					<< "|" << euid 
					<< "||"; 			// circuit number

				std::string channels, operations, address;
				int flag = 0;
				for (auto record : res) {
					if (flag != 0){
						channels.push_back(',');
						operations.push_back(',');
					}

					flag =  (flag == 0) ? 3 : 4;
					channels += record[0].c_str();;
					operations += record[1].c_str(); 
					address = record[2].c_str(); 
				}

				sstr << address << "|" << channels << "||" //channels, device code is empty
					<< flag 
					<< "|||"  // vendor, device type
					<< operations << "##";

				sender(sstr.str());
			}

			if (dtu_flag != 0){
				auto res = w.prepared("get_xhs_dtu")(euid).exec();
				std::stringstream sstr;
				sstr << "@@" << 9 // message type
					<< "|" << 1024; // message sequence number, hard coded, FIXME

				std::string normal_heartbeat, warn_heartbeat, warn_heartbeat_duration;

				std::string signal_type, error_range;
				bool is_first = true;
				for(auto record : res){
					if (!is_first){
						signal_type.push_back(',');
						error_range.push_back(',');
					}
					else 
						is_first = false;

					signal_type += record[0].c_str();
					normal_heartbeat = record[1].c_str();
					warn_heartbeat = record[2].c_str();
					warn_heartbeat_duration = record[3].c_str();
					error_range += record[4].c_str();
				}

				sstr << "|" << normal_heartbeat 
					<< "|" << warn_heartbeat
					<< "|" << warn_heartbeat_duration
					<< "|" << signal_type << "|" << error_range << "##";
				sender(sstr.str());

			}

			w.prepared("clear_xhs_flags")(euid).exec();
		}

	}

	void Database::save_fire_pump_and_fan_info(const info_fire_pump_information& data){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
					w.prepared("update_simid") (data.euid)(data.sim_id)(data.ip_addr_port).exec();
					std::string sql = data.cmd_id == 19? "save_fire_pump_info" 
					: (data.cmd_id == 20? "save_draught_fan_info" : "");
					if (!sql.empty())
						w.prepared(sql)(data.euid) (data.signal)(data.sim_id)
						(data.voltage)(data.ac_det).exec();
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::save_waterater_data(const info_waterater_data& data){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
					w.prepared("update_simid") (data.euid)(data.sim_id)(data.ip_addr_port).exec();
					w.prepared("save_waterater_data")(data.euid) (data.signal)(data.sim_id)
					(data.voltage)(data.velocity)(data.positive_cumulative_flow)
					(data.negative_cumulative_flow)(data.net_cumulative_flow)
					(data.error_code)(data.work_steps)(data.upstream_signal)(data.downstream_signal)
					.exec();
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::save_anolog(const std::string& ip_addr_port, const xhs_send_anolog& data, std::function<void(const std::string&)> sender){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				auto count = std::min(data.channel.size(), data.anolog_value.size());
				count = std::min(count, data.device_status.size());
				uint32_t new_batch = 1;
				for(size_t i = 0; i < count; ++i){
					w.prepared("save_xhs_analog")(new_batch) (data.module)(data.signal)(data.channel[i])
					(data.anolog_value[i])(data.device_status[i])(data.device_addr).exec();
					new_batch = 0;
				}

				check_xhs_flags(w, data.module, sender);

				});
		m_imp->condition_variable.notify_all();
	}

	void Database::save_anolog_config(const std::string& ip_addr_port, const xhs_send_anolog_config& data, std::function<void(const std::string&)> sender){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				std::size_t allsize[] = {
					data.channel.size(),
					data.device_type.size(),
					data.signal_type.size(),
					data.low_value.size(),
					data.high_value.size(),
					data.unit.size()
				};

				auto count = *std::min_element(std::begin(allsize), std::end(allsize));
				uint32_t new_batch = 1;
				for(size_t i = 0; i < count; ++i){
					w.prepared("save_xhs_analog_config") (new_batch)(data.module)(data.channel[i])(data.circuit)
					(data.device_addr)(data.vender)(data.device_type[i])(data.signal_type[i])
					(data.low_value[i])(data.high_value[i])
					(data.unit[i])(data.flag)(data.device_code)
					.exec();
					new_batch = 0;
				}

				check_xhs_flags(w, data.module, sender);
				});
	}

	static std::string to_timestamp(const fire_time& t){
		std::stringstream sstr;
		sstr << (2000 + unsigned(t.year)) << '-' 
			<< std::setfill('0') 
			<< std::setw(2) << unsigned(t.month) << '-'
			<< std::setw(2) << unsigned(t.day) << ' ' 
			<< std::setw(2) << unsigned(t.hour) << ':' 
			<< std::setw(2) << unsigned(t.minute) << ':' 
			<< std::setw(2) << unsigned(t.second);
		return sstr.str();
	}
	template<typename T>
	static std::string to_bit_str(T s){
		return std::bitset<sizeof(T) * 8>(s).to_string();
	}


	bool Database::save_fire_protection(const fire_package& pkg, std::function<void(const std::vector<char>&)> /* sender */){
		buffer_in rd(to_range(pkg.data));
		auto ar = s11n::as_source(rd);
		auto type_flag = load<uint8_t>(ar);
		auto size = load<uint8_t>(ar);

		BOOST_LOG_TRIVIAL(debug) << "DEBUG fire protection: " 
			<< " Seq: " << pkg.seq
			<< " Time: " << to_timestamp(pkg.time)
			<< " Source Address: " << pkg.src_addr
			<< " Dest Address: " << pkg.dest_addr
			<< " Commond: " << uint32_t(pkg.command)
			<< " Type flag: " << uint32_t(type_flag)
			<< " Item count: " << uint32_t(size)
			;


		std::unique_lock<std::mutex> lock(m_imp->mutex);
		for(; size > 0 ; --size) switch (type_flag)
		{
			case 1: 
				{
					info_fire_facility_status item;
					ar & item.sys_type & item.sys_addr & item.status & item.time;
					m_imp->queue.emplace_back([=](pqxx::work& w){
							w.prepared("update_fire_facility_sys_status_state")
							(pkg.src_addr)(pkg.seq)(to_timestamp(item.time))(pkg.dest_addr)
							(uint16_t(item.sys_type))(uint16_t(item.sys_addr))(to_bit_str(item.status))("false")
							.exec();
							});
				}
				break;
			case 2:
				{
					info_fire_part_status item;
					ar & item.sys_type & item.sys_addr & item.part_type & item.part_number & item.part_area & item.status ;
					item.description.resize(31);
					block_read(rd, range<char*>(&*item.description.begin(), &*item.description.begin() + item.description.size()));
					ar & item.time;
					m_imp->queue.emplace_back([=](pqxx::work& w){
							w.prepared("update_fire_facility_part_status_state")
							(pkg.src_addr)(pkg.seq)(to_timestamp(item.time))(pkg.dest_addr)
							(uint16_t(item.sys_type))(uint16_t(item.sys_addr))(uint16_t(item.part_type))(item.part_area)(item.part_number)
							(to_bit_str(item.status))("" /*item.description*/)("false")
							.exec();
							});
				}
			break;
			case 4:
			break;
			case 5:
			break;
			case 6:
			break;
			case 7:
			break;
			case 8:
			break;
			case 21:
			{
				info_fire_device_run_state item;
				ar & item.status & item.time;
				m_imp->queue.emplace_back([=](pqxx::work& w){
						w.prepared("write_fire_device_run_status")
						(pkg.src_addr)(pkg.seq)(to_timestamp(item.time))(pkg.src_addr)(pkg.dest_addr)(to_bit_str(item.status))
						.exec();
#if 0
						w.prepared("update_fire_device_run_status_state")
						(pkg.src_addr)(pkg.seq)(to_timestamp(item.time))(pkg.src_addr)(pkg.dest_addr)(to_bit_str(item.status))
						.exec();
#endif
						});
			}
			break;
			case 24:
			{
				info_fire_device_op_info item;
				ar & item.status & item.operator_id & item.time;
				m_imp->queue.emplace_back([=](pqxx::work& w){
						w.prepared("write_fire_device_op_info")
						(pkg.src_addr)(pkg.seq)(to_timestamp(item.time))(pkg.src_addr)(pkg.dest_addr)(uint16_t(item.operator_id))(to_bit_str(item.status))
						.exec();
#if 0
						w.prepared("update_fire_device_op_info_state")
						(pkg.src_addr)(pkg.seq)(to_timestamp(item.time))(pkg.src_addr)(pkg.dest_addr)(uint16_t(item.operator_id))(to_bit_str(item.status))
						.exec();
#endif
						});
			}
			break;
			case 25:
			break;
			case 26:
			break;
			case 28:
			break;
			case 128:
			break;
			case 129:
			break;
			case 130:
			break;
			case 131:
			break;
			case 132:
			break;
			case 133:
			break;
			case 134:
			{
				info_fire_facility_status item;
				ar & item.sys_type & item.sys_addr & item.status & item.time;
				m_imp->queue.emplace_back([=](pqxx::work& w){
						w.prepared("update_fire_facility_sys_status_state")
						(pkg.src_addr)(pkg.seq)(to_timestamp(item.time))(pkg.dest_addr)
						(uint16_t(item.sys_type))(uint16_t(item.sys_addr))(to_bit_str(item.status))("true")
						.exec();
						});
			}
			break;
			case 135:
			{
				info_fire_part_status item;
				ar & item.sys_type & item.sys_addr & item.part_type & item.part_number & item.part_area & item.status ;
				ar & item.time;
				m_imp->queue.emplace_back([=](pqxx::work& w){
						w.prepared("update_fire_facility_part_status_state")
						(pkg.src_addr)(pkg.seq)(to_timestamp(item.time))(pkg.dest_addr)
						(uint16_t(item.sys_type))(uint16_t(item.sys_addr))(uint16_t(item.part_type))(item.part_area)(item.part_number)
						(to_bit_str(item.status))("" /* item.description */ )("true")
						.exec();
						});
			}
			break;
			case 136:
			{
				info_fire_device_run_state item;
				ar & item.status & item.time;
				m_imp->queue.emplace_back([=](pqxx::work& w){
						w.prepared("update_fire_device_run_status_state")
						(pkg.src_addr)(pkg.seq)(to_timestamp(item.time))(pkg.src_addr)(pkg.dest_addr)(to_bit_str(item.status))
						.exec();
						});
			}
			break;
			default:
			break;
		}
		m_imp->condition_variable.notify_all();
		return true;
	}
	void Database::voiceHeartbeat(const std::string& ip_addr_port, const std::string& euid, uint16_t signal, const std::string& sim_id, uint16_t voice, std::function<void(uint8_t )> sender){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("write_voiceHB") (euid)(signal)(sim_id)(voice).exec();
				w.prepared("update_simid") (euid)(sim_id)(ip_addr_port).exec();
				auto result = w.prepared("get_voice_id") (euid)(voice).exec();
				if (!result.empty()){
					BOOST_LOG_TRIVIAL(debug) << "DEBUG send horn: " << euid;
					auto command = boost::lexical_cast<uint16_t>(result.begin()->begin()->c_str());
					sender(uint8_t(command));
				}

				});
		m_imp->condition_variable.notify_all();

	}
	void Database::clear_device_init_flag(const std::string& euid){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("clear_device_init_flag") (euid).exec();
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::clear_device_param_flag(const std::string& euid){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("clear_device_param_flag") (euid).exec();
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::forwardData(const std::string& euid, std::function<void(const std::string&, const std::string&)> sender){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				auto result = w.prepared("get_forward_endpoint") (euid).exec();
				if (!result.empty()){
					auto field_itr = result.begin()->begin();
					std::string ip = field_itr->c_str();
					++field_itr;
					std::string port = field_itr->c_str();
					if (!ip.empty() && !port.empty()){
					BOOST_LOG_TRIVIAL(debug) << "forward to " << ip << ":" << port;
					sender(ip, port);
					}
				}
				});
		m_imp->condition_variable.notify_all();
	}

	void Database::startWorker(){
		if (!m_imp->workerIsRunning){
			m_imp->workerIsRunning = true;
			m_imp->worker = std::thread([&]{m_imp->do_work();});
		}

	}
	bool Database::workerIsRunning() const{
		return m_imp->workerIsRunning;
	}
	void Database::stopWorker(){
		if (m_imp->workerIsRunning){
			m_imp->workerIsRunning = false;
			if (m_imp->worker.joinable()){
				m_imp->condition_variable.notify_all();
				m_imp->worker.join();
			}
		}
	}

	void send_sponge_response(pqxx::work& w, const std::string& euid, std::function<void(const std::vector<char>&)> sender){
		std::vector<char> response;
		buffer_out wr(response);
		auto ar = s11n::as_sink(wr);
		ar & K_signature & K_version & K_reserved 
			& uint16_t(0x16);

		auto result = w.prepared("get_spongecity_downconfig")(euid).exec();

		if (!result.empty()){
			auto field_itr = result.begin()->begin();
			auto status = field_itr->as<int>();
			if (status != 0){
				ar & uint16_t(35)		// total size
					& uint16_t(0x3a)	// SysTimeCon_ID
					;
				for(int i = 0; i < 4; ++i){
					++field_itr;
					ar & uint16_t(field_itr->as<int>());
				}
				ar & uint16_t(0x3d);
				for(int i = 0; i < 4; ++i){
					++field_itr;
					ar & uint16_t(field_itr->as<int>());
				}
				ar & uint16_t(0x3b);
				for(int i = 0; i < 2; ++i){
					++field_itr;
					ar & uint16_t(field_itr->as<int>());
				}
				ar & uint16_t(0x3c);
				for(int i = 0; i < 2; ++i){
					++field_itr;
					ar & uint16_t(field_itr->as<int>());
				}
				ar & uint16_t(0x50);
				++field_itr;
				ar & uint8_t(field_itr->as<int>());


				sender(response);
				w.prepared("clear_spongecity_downconfig")(euid).exec();
				return;
			}
		}

		ar & uint16_t(3)
			& uint16_t(0x3e)
			& uint8_t(1)
			;
		sender(response);

	}
	int high_of(uint16_t v){
		return (v & 0xff00) >> 8;
	}
	int low_of(uint16_t v){
		return (v & 0xff);
	}
	void Database::save_sponge_data(const sponge_data& data, std::function<void(const std::vector<char>&)> sender){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
#if 0
				w.prepared("update_simid") (data.euid)(data.sim_id)(data.ip_addr_port).exec();
				w.prepared("save_spongecity_data") (data.euid)
				(data.signal)(data.sim_id)(data.voltage)
				(high_of(data.sensor_cond[0])) (low_of(data.sensor_cond[0]))
				(high_of(data.sensor_cond[1])) (low_of(data.sensor_cond[1]))
				(high_of(data.sensor_cond[2])) (low_of(data.sensor_cond[2]))
				(high_of(data.sensor_cond[3])) (low_of(data.sensor_cond[3]))
				(data.flow.flow) (data.flow.flow_all) (data.flow.temperature) (data.flow.rate) (data.flow.depth)
				(data.ss.ss) (data.ss.factor, !std::isnan(data.ss.factor)) (data.ss.humidity)
				(data.cod.cod) (data.cod.turbidity)
				(data.rain.hyetal)(data.rain.hyetal_all)
				.exec();
#endif

				send_sponge_response(w, data.euid, sender);
				});
		m_imp->condition_variable.notify_all();

	}
	void Database::save_sponge_upconfig(const sponge_upconfig& data, std::function<void(const std::vector<char>&)> sender){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
#if 0
				w.prepared("update_simid") (data.euid)(data.sim_id)(data.ip_addr_port).exec();
				w.prepared("save_spongecity_upconfig") (data.euid)
				(data.sys_time_cond[0]) (data.sys_time_cond[1]) (data.sys_time_cond[2]) (data.sys_time_cond[3])
				(data.flow_cond[0]) (data.flow_cond[1]) (data.cod_cond[0]) (data.cod_cond[1])(data.flow_cal_con)
				.exec();
#endif
				
				send_sponge_response(w, data.euid, sender);
				});
		m_imp->condition_variable.notify_all();
	}
	void Database::saveSoak(const std::string& ip_addr_port, const std::string& euid, int32_t soak, int32_t trend, 
			uint16_t signal, const std::string& sim_id, int32_t voltage){
		std::unique_lock<std::mutex> lock(m_imp->mutex);
		m_imp->queue.emplace_back([=](pqxx::work& w){
				w.prepared("update_simid")(euid)(sim_id)(ip_addr_port).exec();
				w.prepared("write_soak") (euid)(signal)(sim_id)(voltage)(soak).exec();
				});
		m_imp->condition_variable.notify_all();
	}
}

